package com.yxc.ejad.utils;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.mapping.FieldType;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch.core.*;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.rest_client.RestClientTransport;
import co.elastic.clients.util.ObjectBuilder;
import com.alibaba.fastjson.JSONObject;
import com.yxc.ejad.annotation.ElasticsearchFieldConfig;
import com.yxc.ejad.annotation.ElasticsearchIndexConfig;
import com.yxc.ejad.vo.PageVo;
import com.yxc.ejad.vo.ScrollListVo;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.conn.ssl.NoopHostnameVerifier;
import org.apache.http.message.BasicHeader;
import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
import org.apache.http.ssl.SSLContextBuilder;
import org.elasticsearch.client.RestClient;

import javax.net.ssl.SSLContext;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.time.LocalDateTime;
import java.util.List;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * ElasticsearchClient的工具类
 *
 * @author: YXC
 * @time: 2023/9/28 10:27
 **/
public class ElasticsearchClientUtils {

    public static final String SETTINGS = "settings";
    public static final String PROPERTIES = "properties";
    public static final String MAPPINGS = "mappings";
    public static final String TYPE = "type";
    public static final String INDEX = "index";
    public static final String NUMBER_OF_SHARDS = "number_of_shards";
    public static final String NUMBER_OF_REPLICAS = "number_of_replicas";

    /**
     * 获取es客户端
     *
     * @param host:   ip
     * @param port:   端口
     * @param apiKey: apikey  NHFicFZvd0JzTTQ1NjhLOS1pTTc6ZW02TFFnT2VUX2VxUE9abl9VczAwZw==
     * @return co.elastic.clients.elasticsearch.ElasticsearchClient
     * @author YXC
     * @date 2023/12/11 11:29
     **/
    public static ElasticsearchClient getClient(String host, Integer port, String apiKey) throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException {

        // 设置SSL
        SSLContext sslContext = new SSLContextBuilder().loadTrustMaterial(null, (x509Certificates, s) -> true).build();
        SSLIOSessionStrategy sessionStrategy = new SSLIOSessionStrategy(sslContext, NoopHostnameVerifier.INSTANCE);

        // Create the low-level client
        RestClient restClient = RestClient
                .builder(HttpHost.create("https://" + host + ":" + port))
                .setDefaultHeaders(new Header[]{
                        new BasicHeader("Authorization", "ApiKey " + apiKey)
                })
                .setHttpClientConfigCallback(
                        b -> b.setSSLStrategy(sessionStrategy)
                )
                .build();

        // Create the transport with a Jackson mapper
        ElasticsearchTransport transport = new RestClientTransport(
                restClient, new JacksonJsonpMapper());

        // And create the API client
        ElasticsearchClient esClient = new ElasticsearchClient(transport);
        return esClient;
    }

//    public static void main(String[] args) throws NoSuchAlgorithmException, KeyStoreException, KeyManagementException, IOException {
//        ElasticsearchClient client = getClient("10.26.81.47", 9200, "OFEzZ2NZMEJvdHg0WTZqLTFlMHc6UWVIclhkOEpSSmFoTmtPMFd6QVRZUQ==");
//        Boolean test_create = createIndex("test_create", BinTable.class, client);
//        System.out.println(test_create);
//    }

    /**
     * 判断索引是否存在
     *
     * @param indexName: 索引名
     * @param client:    es客户端
     * @return java.lang.Boolean
     * @author YXC
     * @date 2023/12/11 11:33
     **/
    public static Boolean indexExist(String indexName, ElasticsearchClient client) throws IOException {
        if (indexName == null || "".equals(indexName)) {
            return false;
        }
        return client.indices().exists(i -> i.index(indexName)).value();
    }

    /**
     * 创建索引
     *
     * @param indexName:
     * @param clazz:
     * @param client:
     * @return java.lang.Boolean
     * @author YXC
     * @date 2024/2/4 10:15
     **/
    public static Boolean createIndex(String indexName, Class<?> clazz, ElasticsearchClient client) throws IOException {
        // 判断是否存在
        if (indexExist(indexName, client)) {
            throw new RuntimeException("索引" + indexName + "已经存在");
        }
        JSONObject indexJson = getIndexJson(clazz);
        ByteArrayInputStream inputStream = new ByteArrayInputStream(indexJson.toJSONString().getBytes());
        CreateIndexResponse response = client.indices().create(i -> i.index(indexName).withJson(inputStream));
        return response.acknowledged();
    }

    /**
     * 获取类的json配置
     *
     * @param clazz:
     * @return com.alibaba.fastjson.JSONObject
     * @author YXC
     * @date 2024/2/4 10:03
     **/
    public static JSONObject getIndexJson(Class<?> clazz) {
        if (clazz == null) {
            return new JSONObject();
        }
        ElasticsearchIndexConfig indexConfig = clazz.getDeclaredAnnotation(ElasticsearchIndexConfig.class);
        String indexName = getIndexName(indexConfig, clazz);

        JSONObject indexJson = new JSONObject();
        JSONObject mapping = getMapping(clazz);
        // setting字段
        JSONObject setting = new JSONObject();
        if (indexConfig != null) {
            setting.put(ElasticsearchClientUtils.NUMBER_OF_SHARDS, indexConfig.numberOfShards());
            setting.put(ElasticsearchClientUtils.NUMBER_OF_REPLICAS, indexConfig.numberOfReplicas());
        }
        indexJson.put(SETTINGS, setting);
        indexJson.put(MAPPINGS, mapping);
        // TODO: 2024/2/2 未完成
        return indexJson;
    }

    /**
     * 获取字段的mapping
     *
     * @param clazz:
     * @return com.alibaba.fastjson.JSONObject
     * @author YXC
     * @date 2024/2/2 16:20
     **/
    public static JSONObject getMapping(Class<?> clazz) {

        JSONObject mapping = new JSONObject();
        JSONObject properties = new JSONObject();
        if (clazz == null) {
            return mapping;
        }
        for (Field field : clazz.getDeclaredFields()) {
            JSONObject fieldJson = new JSONObject();

            ElasticsearchFieldConfig fieldConfig = field.getDeclaredAnnotation(ElasticsearchFieldConfig.class);
            if (Modifier.isStatic(field.getModifiers()) || (fieldConfig != null && !fieldConfig.exist())) {
                continue;
            }
            // 字段名称
            String fieldName = getFieldName(fieldConfig, field);
            // 判断类型
            String type = getFieldType(fieldConfig, field);
            // 是否建立索引
            boolean index = fieldConfig == null || fieldConfig.index();

            fieldJson.put(ElasticsearchClientUtils.TYPE, type);
            fieldJson.put(ElasticsearchClientUtils.INDEX, index);

            properties.put(fieldName, fieldJson);
        }
        mapping.put(ElasticsearchClientUtils.PROPERTIES, properties);
        return mapping;
    }

    /**
     * 获取字段类型
     * 未知类型使用text
     *
     * @param fieldConfig: 字段配置
     * @param field:       字段
     * @return java.lang.String
     * @author YXC
     * @date 2024/2/2 15:56
     **/
    public static String getFieldType(ElasticsearchFieldConfig fieldConfig, Field field) {
        if (field == null) {
            throw new NullPointerException("field must be not null");
        }
        if (fieldConfig != null && fieldConfig.type() != null) {
            return fieldConfig.type().jsonValue();
        }
        Class<?> fieldType = field.getType();
        if (String.class.equals(fieldType)) {
            return FieldType.Text.jsonValue();
        }
        if (fieldType.equals(Integer.class)) {
            return FieldType.Integer.jsonValue();
        }
        if (fieldType.equals(Long.class)) {
            return FieldType.Long.jsonValue();
        }
        if (fieldType.equals(Double.class)) {
            return FieldType.Double.jsonValue();
        }
        if (fieldType.equals(LocalDateTime.class)) {
            return FieldType.Date.jsonValue();
        }
        return FieldType.Text.jsonValue();
    }

    /**
     * 获取索引名
     *
     * @param indexConfig: 自定义的索引配置
     * @param clazz:       类
     * @return java.lang.String
     * @author YXC
     * @date 2024/2/2 15:24
     **/
    public static String getIndexName(ElasticsearchIndexConfig indexConfig, Class<?> clazz) {
        if (clazz == null) {
            throw new NullPointerException("clazz must be not null");
        }
        if (indexConfig == null || "".equals(indexConfig.indexName())) {
            return clazz.getName();
        }
        return indexConfig.indexName();
    }

    /**
     * 获取字段名
     *
     * @param fieldConfig: 字段配置
     * @param field:       字段
     * @return java.lang.String
     * @author YXC
     * @date 2024/2/2 15:45
     **/
    private static String getFieldName(ElasticsearchFieldConfig fieldConfig, Field field) {
        if (field == null) {
            throw new NullPointerException("field must be not null");
        }
        if (fieldConfig == null || "".equals(fieldConfig.fieldName())) {
            return field.getName();
        }
        return fieldConfig.fieldName();
    }

    /**
     * 添加一条数据
     *
     * @param entity:    数据实体类
     * @param indexName: 索引名称
     * @param client:    es客户端
     * @return int
     * @author YXC
     * @date 2023/12/11 15:08
     **/
    public static <T> int saveOne(T entity, String indexName, ElasticsearchClient client) throws IOException {
        IndexResponse indexResponse = client.index(i ->
                i.index(indexName)
                        .document(entity)
        );
        return 1;
    }

    /**
     * 添加一条数据，指定id
     *
     * @param entity:    要添加的数据
     * @param id:        id
     * @param indexName: 索引名称
     * @param client:    es客户端
     * @return int
     * @author YXC
     * @date 2023/12/11 15:10
     **/
    public static <T> int saveOne(T entity, String id, String indexName, ElasticsearchClient client) throws IOException {
        IndexResponse index = client.index(
                i -> i.index(indexName)
                        .id(id)
                        .document(entity)
        );
        return 1;
    }

    /**
     * 批量存储数据
     *
     * @param dataList:  数据列表
     * @param indexName: 索引名
     * @param client:    es客户端
     * @return int
     * @author YXC
     * @date 2023/12/11 15:26
     **/
    public static <T> int saveBatch(List<T> dataList, String indexName, ElasticsearchClient client) throws IOException {
        BulkRequest.Builder builder = new BulkRequest.Builder();
        dataList.forEach(
                data -> builder.operations(
                        op -> op.index(
                                i -> i.index(indexName)
                                        .document(data)
                        )
                )
        );
        BulkResponse bulkResponse = client.bulk(builder.build());
        return dataList.size();
    }

    /**
     * 根据搜索条件删除数据
     *
     * @param query:  搜索条件
     * @param client: es客户端
     * @return java.lang.Long
     * @author YXC
     * @date 2024/3/12 15:35
     **/
    public static <T> Long deleteByQuery(Function<Query.Builder, ObjectBuilder<Query>> query,
                                         ElasticsearchClient client) throws IOException {
        DeleteByQueryResponse response = client.deleteByQuery(
                d -> d.query(query)
        );
        return response.deleted();
    }

    /**
     * 分页查询
     *
     * @param index:          索引
     * @param queryFn:        查询函数
     * @param pageNo:         第几页
     * @param pageSize:       每页多少条
     * @param trackTotalHits: 是否显示真实的总数
     * @param sortField:      要排序的字段
     * @param order:          正序还是反序
     * @param clazz:          查询对应的java类
     * @param client:         es客户端
     * @return com.yxc.ejad.vo.PageVo<T>
     * @author YXC
     * @date 2023/12/14 16:09
     **/
    public static <T> PageVo<T> listByQueryPageable(
            String index,
            Function<Query.Builder, ObjectBuilder<Query>> queryFn,
            Integer pageNo,
            Integer pageSize,
            Boolean trackTotalHits,
            String sortField,
            SortOrder order,
            Class<T> clazz,
            ElasticsearchClient client
    ) throws IOException {
        Function<SearchRequest.Builder, ObjectBuilder<SearchRequest>> searchFn = s -> {
            s.index(index);
            if (queryFn != null) {
                s.query(queryFn);
            }
            if (checkPageParams(pageNo, pageSize)) {
                s.from((pageNo - 1) * pageSize);
                s.size(pageSize);
            }
            s.trackTotalHits(t -> t.enabled(trackTotalHits));
            if (sortField != null && "".equals(sortField) && order != null) {
                s.sort(so -> so.field(f -> f.field(sortField).order(order)));
            }
            return s;
        };

        if (checkPageParams(pageNo, pageSize)) {

        }
        SearchResponse<T> searchResponse = client.search(searchFn, clazz);
        List<T> dataList = searchResponse.hits().hits().stream().map(Hit::source).collect(Collectors.toList());

        PageVo<T> pageVo = new PageVo<>();
        pageVo.setDataList(dataList);
        if (pageNo == null || pageSize == null) {
            return pageVo;
        }
        pageVo.setPageNo(pageNo);
        pageVo.setPageSize(pageSize);
        pageVo.setTotalCount(searchResponse.hits().total().value());
        pageVo.setIsReal(trackTotalHits);
        return pageVo;
    }

    /**
     * 分页查询全部
     *
     * @param pageNo:         第几页
     * @param index:          索引
     * @param pageSize:       每页多少条
     * @param trackTotalHits: 是否返回全量总数
     * @param clazz:          要返回的java类
     * @param client:         esClient
     * @return cas.iie.xdtest.vo.PageVo<T>
     * @author YXC
     * @date 2024/1/16 15:11
     **/
    public static <T> PageVo<T> listPageable(
            String index,
            Integer pageNo,
            Integer pageSize,
            Boolean trackTotalHits,
            Class<T> clazz,
            ElasticsearchClient client
    ) throws IOException {
        return ElasticsearchClientUtils.listByQueryPageable(index, null, pageNo, pageSize, trackTotalHits, null, null, clazz, client);
    }

    /**
     * 分页查询全部
     *
     * @param index:          索引名
     * @param pageNo:         第几页
     * @param pageSize:       每页多少条
     * @param trackTotalHits: 是否返回全部总数
     * @param sortField:      排序字段
     * @param order:          排序方式
     * @param clazz:          java类
     * @param client:         esClient
     * @return cas.iie.xdtest.vo.PageVo<T>
     * @author YXC
     * @date 2024/1/16 15:38
     **/
    public static <T> PageVo<T> listPageable(
            String index,
            Integer pageNo,
            Integer pageSize,
            Boolean trackTotalHits,
            String sortField,
            SortOrder order,
            Class<T> clazz,
            ElasticsearchClient client
    ) throws IOException {
        return ElasticsearchClientUtils.listByQueryPageable(index, null, pageNo, pageSize, trackTotalHits, sortField, order == null ? SortOrder.Desc : order, clazz, client);
    }

    /**
     * 分页条件搜索，没有排序
     *
     * @param index:          索引
     * @param queryFn:        搜索函数
     * @param pageNo:         第几页
     * @param pageSize:       每页多少条
     * @param trackTotalHits: 是否显示总页数
     * @param clazz:          搜索的类
     * @param client:         esClient
     * @return cas.iie.xdtest.vo.PageVo<T>
     * @author YXC
     * @date 2024/1/16 15:21
     **/
    public static <T> PageVo<T> listByQueryPageable(
            String index,
            Function<Query.Builder, ObjectBuilder<Query>> queryFn,
            Integer pageNo,
            Integer pageSize,
            Boolean trackTotalHits,
            Class<T> clazz,
            ElasticsearchClient client
    ) throws IOException {
        return ElasticsearchClientUtils.listByQueryPageable(index, queryFn, pageNo, pageSize, trackTotalHits, null, null, clazz, client);
    }

    /**
     * 条件搜索，没有分页，没有排序
     *
     * @param index:   索引名
     * @param queryFn: 搜索函数
     * @param clazz:   搜索的java类
     * @param client:  客户端
     * @return List<T>
     * @author YXC
     * @date 2023/12/14 16:17
     **/
    public static <T> List<T> listByQuery(
            String index,
            Function<Query.Builder, ObjectBuilder<Query>> queryFn,
            Class<T> clazz,
            ElasticsearchClient client
    ) throws IOException {
        return ElasticsearchClientUtils.listByQuery(index, 100, queryFn, null, null, clazz, client);
    }

    /**
     * 根据条件搜索
     *
     * @param index:
     * @param size:
     * @param queryFn:
     * @param clazz:
     * @param client:
     * @return java.util.List<T>
     * @author YXC
     * @date 2024/3/13 16:55:15
     **/
    public static <T> List<T> listByQuery(
            String index,
            Integer size,
            Function<Query.Builder, ObjectBuilder<Query>> queryFn,
            Class<T> clazz,
            ElasticsearchClient client
    ) throws IOException {
        return ElasticsearchClientUtils.listByQuery(index, size, queryFn, null, null, clazz, client);
    }

    /**
     * 根据条件搜索
     *
     * @param index:     索引名
     * @param size:      返回数量
     * @param queryFn:   搜索函数
     * @param sortField: 排序字段
     * @param order:     排序
     * @param clazz:     类
     * @param client:    客户端
     * @return java.util.List<T>
     * @author YXC
     * @date 2024/3/13 16:54:11
     **/
    public static <T> List<T> listByQuery(
            String index,
            Integer size,
            Function<Query.Builder, ObjectBuilder<Query>> queryFn,
            String sortField,
            SortOrder order,
            Class<T> clazz,
            ElasticsearchClient client
    ) throws IOException {
        Function<SearchRequest.Builder, ObjectBuilder<SearchRequest>> searchFn = s -> {
            s.index(index);
            if (queryFn != null) {
                s.query(queryFn);
            }
            s.size(size);
            if (sortField != null && "".equals(sortField) && order != null) {
                s.sort(so -> so.field(f -> f.field(sortField).order(order)));
            }
            return s;
        };

        SearchResponse<T> searchResponse = client.search(searchFn, clazz);
        List<T> dataList = searchResponse.hits().hits().stream().map(Hit::source).collect(Collectors.toList());

        return dataList;
    }

    /**
     * 按条件查询，游标查询
     *
     * @param index:     查询的索引
     * @param queryFn:   查询函数
     * @param size:      每次的数量
     * @param time:      游标有效时间，比如：1m, 10s, 1h
     * @param sortField: 排序字段
     * @param order:     排序方式
     * @param clazz:     对应java类
     * @param client:    esClient
     * @return com.yxc.ejad.vo.ScrollListVo<T>
     * @author YXC
     * @date 2024/1/19 16:11
     **/
    public static <T> ScrollListVo<T> listByQueryScroll(
            String index,
            Function<Query.Builder, ObjectBuilder<Query>> queryFn,
            Integer size,
            String time,
            String sortField,
            SortOrder order,
            Class<T> clazz,
            ElasticsearchClient client
    ) throws IOException {
        Function<SearchRequest.Builder, ObjectBuilder<SearchRequest>> searchFn = s -> {
            s.index(index);
            if (time != null && "".equals(time)) {
                s.scroll(t -> t.time(time));
            } else {
                s.scroll(t -> t.time("1m"));
            }

            if (queryFn != null) {
                s.query(queryFn);
            }
            if (size != null && size != 0) {
                s.size(size);
            }
            if (sortField != null && "".equals(sortField) && order != null) {
                s.sort(so -> so.field(f -> f.field(sortField).order(order)));
            }
            return s;
        };
        SearchResponse<T> searchResponse = client.search(searchFn, clazz);
        List<T> dataList = searchResponse.hits().hits().stream().map(Hit::source).collect(Collectors.toList());
        String scrollId = searchResponse.scrollId();
        return new ScrollListVo<>(dataList, scrollId, time);
    }

    /**
     * 游标搜索
     *
     * @param scrollId: 游标id
     * @param time:     游标有效时间
     * @param clazz:    搜索类
     * @param client:   esClient
     * @return com.yxc.ejad.vo.ScrollListVo<T>
     * @author YXC
     * @date 2024/1/19 16:12
     **/
    public static <T> ScrollListVo<T> scroll(
            String scrollId,
            String time,
            Class<T> clazz,
            ElasticsearchClient client
    ) throws IOException {
        ScrollResponse<T> scrollResponse = client.scroll(s -> s.scrollId(scrollId).scroll(t -> t.time(time)), clazz);
        String newScrollId = scrollResponse.scrollId();
        List<T> dataList = scrollResponse.hits().hits().stream().map(Hit::source).collect(Collectors.toList());
        return new ScrollListVo<>(dataList, newScrollId, time);
    }

    private static Boolean checkPageParams(Integer pageNo, Integer pageSize) {
        return pageNo != null && pageSize != null && pageNo > 0 && pageSize > 0;
    }

}
